package sagroup.dracots.policy;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Logger;

import lime2.Location;
import sagroup.dracots.TupleType;

/**
 * PolicyRegistrar maintains the replication policy info of different tuple
 * types on different host.
 * 
 * @author Huang Xiaodong
 */
public class PolicyRegistrar {
	/**
	 * HashMap that keeps the replication policy info. A two-level HashMap, that
	 * is, HashMap<tupleType, HashMap<replicaLocation, replicaPolicy>>.
	 */
	private HashMap<TupleType, HashMap<Location, ReplicationPolicy>> policyMap;
	
	private Logger LOG = Logger.getLogger(this.getClass().getName());

	/**
	 * Constructor
	 */
	public PolicyRegistrar() {
		policyMap = new HashMap<TupleType, HashMap<Location, ReplicationPolicy>>();
	}

	/**
	 * Show the contents of PolicyRegistrar.
	 * 
	 * @author Huang Xiaodong
	 */
	public void ShowAll() {
		System.out.println("Show the contents of replicaRegistra.");
		for (TupleType type : policyMap.keySet()) {
			System.out.println("Type = " + type + " ,");
			HashMap<Location, ReplicationPolicy> inner = policyMap.get(type);
			if (inner != null) {
				for (Location loc : inner.keySet())
					System.out.println("   Location = " + loc
							+ " and Replication Policy = " + inner.get(loc));
			}
		}
	}

	/*
	 * Return the whole hash Map of replica policy
	 * 
	 * @author Jinxue Zhang
	 */
	public HashMap<TupleType, HashMap<Location, ReplicationPolicy>> getAllReplicaPolicy(){
		return policyMap;
	}
	
	public String toString() {
		return "policyMap = " + policyMap.toString();
	}

	/**
	 * Register a specified replication policy.
	 * 
	 * @param targetType
	 * @param targetNode
	 * @param policy
	 * @author Huang Xiaodong
	 */
	public void registerReplicaPolicy(TupleType targetType,
			Location targetNode, ReplicationPolicy policy) {
		HashMap<Location, ReplicationPolicy> inner = policyMap.get(targetType);
		if (inner == null) {
			inner = new HashMap<Location, ReplicationPolicy>();
			policyMap.put(targetType, inner);
		}
		ReplicationPolicy oldPolicy = inner.get(targetNode);
		if (oldPolicy == null)
			inner.put(targetNode, policy);
		else
			// Already registered
			updateReplicaPolicy(targetType, targetNode, policy);
	}

	/**
	 * Deregister a replication policy with specified target type and target
	 * node.
	 * 
	 * @param targetType
	 * @param targetNode
	 * @author Huang Xiaodong
	 */
	public void deregisterReplicaPolicy(TupleType targetType,
			Location targetNode) {
		HashMap<Location, ReplicationPolicy> inner = policyMap.get(targetType);
		if (inner != null) {
			inner.remove(targetNode);
			if (inner.isEmpty())
				policyMap.remove(targetType);
		}
	}

	/**
	 * Deregister replications policy with specified type.
	 * 
	 * @param targetType
	 * @return
	 * @author Huang Xiaodong
	 */
	public Location[] deregisterReplicaPolicy(TupleType targetType) {
		HashMap<Location, ReplicationPolicy> inner = policyMap
				.remove(targetType);
		if (inner != null && inner.keySet() != null) {
			Object[] obj_loc = inner.keySet().toArray();
			Location[] ret = new Location[obj_loc.length];
			for (int i = 0; i < obj_loc.length; ++i)
				ret[i] = (Location) obj_loc[i];
			return ret;
		} else
			return null;
	}

	/**
	 * Update the policy of a replication policy with specified targetType and
	 * targetNode.
	 * 
	 * @param targetType
	 * @param targetNode
	 * @param policy
	 * @author Huang Xiaodong
	 */
	public void updateReplicaPolicy(TupleType targetType, Location targetNode,
			ReplicationPolicy policy) {
		HashMap<Location, ReplicationPolicy> inner = policyMap.get(targetType);
		if (inner == null || inner.get(targetNode) == null)
			registerReplicaPolicy(targetType, targetNode, policy);
		else {
			inner.remove(targetNode);
			inner.put(targetNode, policy);
		}
	}

	/**
	 * Update the policy of a replication policy with specified targetType.
	 * 
	 * @param targetType
	 * @param targetNode
	 * @param policy
	 * @author Huang Xiaodong
	 */
	public Location[] updateReplicaPolicy(TupleType targetType,
			ReplicationPolicy newPolicy) {
		HashMap<Location, ReplicationPolicy> inner = policyMap.get(targetType);
		Location[] ret = null;
		if (inner != null) {
			if (inner.isEmpty()) {
				policyMap.remove(targetType);
			} else {
				if (inner.keySet() != null) {
					Object[] obj_loc = inner.keySet().toArray();
					ret = new Location[obj_loc.length];
					for (int i = 0; i < obj_loc.length; ++i)
						ret[i] = (Location) obj_loc[i];
					for (Location targetNode : inner.keySet()) {
						inner.remove(targetNode);
						inner.put(targetNode, newPolicy);
					}
				}
			}
		}
		return ret;
	}

	public ReplicationPolicy getReplicaPolicy(TupleType targetType,
			Location targetNode) {
		HashMap inner = policyMap.get(targetType);
		if (inner == null)
			return null;
		else
			return (ReplicationPolicy) inner.get(targetNode);
	}
	
	/**
	 * Give all the hosts that register FR for the specified type
	 * @param targetType
	 * @return a host array. If no host register FR for that type
	 * , return an empty array.
	 * @author Xing Jiankuan
	 */
	public Location[] getFRHosts(TupleType targetType) {
		HashMap inner = policyMap.get(targetType);
		if(inner == null) return new Location[0];
		List hostsWithFR = new ArrayList();
		Iterator it = inner.keySet().iterator();
		while(it.hasNext()) {
			Object targetNode = it.next();
			ReplicationPolicy policy = (ReplicationPolicy)inner.get(targetNode);
			if(policy.equals(ReplicationPolicy.FR))
				hostsWithFR.add(targetNode);
		}
		Location[] ret = new Location[hostsWithFR.size()];
		for(int i = 0; i < ret.length; i++) {
			ret[i] = (Location)hostsWithFR.get(i);
		}
		return ret;
	}
	
	/**
	 * Give all the hosts that register FR for the specified type
	 * except <code>Location.HERE</code>
	 * @param targetType
	 * @return a host array. If no host register FR for that type
	 * , return an empty array.
	 * @author Xing Jiankuan
	 */
	public Location[] getFRHostsWithoutHere(TupleType targetType) {
		HashMap<Location, ReplicationPolicy> inner = policyMap.get(targetType);
		if(inner == null) return new Location[0];
		List<Location> hostsWithFR = new ArrayList<Location>();
		Iterator<Location> it = inner.keySet().iterator();
		while(it.hasNext()) {
			Location targetNode = it.next();
			if(targetNode.equals(Location.HERE))
				continue; //ignore here
			ReplicationPolicy policy = (ReplicationPolicy)inner.get(targetNode);
			if(policy.equals(ReplicationPolicy.FR))
				hostsWithFR.add(targetNode);
		}
		Location[] ret = new Location[hostsWithFR.size()];
		for(int i = 0; i < ret.length; i++) {
			ret[i] = (Location)hostsWithFR.get(i);
		}
		return ret;
	}
}